Spark读取和存储HDFS上的数据 |
您所在的位置:网站首页 › spark 保存文件 › Spark读取和存储HDFS上的数据 |
本篇来介绍一下通过Spark来读取和HDFS上的数据,主要包含四方面的内容:将RDD写入HDFS、读取HDFS上的文件、将HDFS上的文件添加到Driver、判断HDFS上文件路径是否存在。 本文的代码均在本地测试通过,实用的环境时MAC上安装的Spark本地环境。 1、启动Hadoop 首先启动咱们的Hadoop,在hadoop的目录下执行下面的命令: rm -rf tmp mkdir tmp cd sbin hadoop namenode -format start-dfs.sh start-yarn.sh查看是否启动成功: 2、将RDD写入HDFS先创建一个SparkSession: val spark = SparkSession .builder() .appName("Spark SQL basic example") .enableHiveSupport() .getOrCreate() spark.sparkContext.setLogLevel("WARN")将RDD写入HDFS使用的函数是saveAsTextFile: val modelNames = Array("FM","FFM","DEEPFM","NFM","DIN","DIEN") val modelNamesRdd = spark.sparkContext.parallelize(modelNames,1) modelNamesRdd.saveAsTextFile("hdfs://localhost:9000/user/root/modelNames")接下来,我们查看一下是否保存成功: 可以看到RDD在HDFS上是分块存储的,由于我们只有一个分区,所以只有part-0000。假设我们存储一个包含两个分区的RDD: val modelNames3 = Array("FM","FFM","DEEPFM","NFM","DIN","DIEN") val modelNames3Rdd = spark.sparkContext.parallelize(modelNames3,2) modelNames3Rdd.saveAsTextFile("hdfs://localhost:9000/user/root/modelNames3")再次查看,可以看到有part-00000和part-00001: 3、读取HDFS上的文件读取HDFS上的文件,使用textFile方法: val modelNames2 = spark.sparkContext.textFile("hdfs://localhost:9000/user/root/modelNames/part-00000") val modelNames4 = spark.sparkContext.textFile("hdfs://localhost:9000/user/root/modelNames3/")读取时是否加最后的part-00000都是可以的,当只想读取某个part,则必须加上。 4、将HDFS上的文件添加到Driver有时候,我们并不想直接读取HDFS上的文件,而是想对应的文件添加到Driver上,然后使用java或者Scala的I/O方法进行读取,此时使用addFile和get方法来实现: val files = "hdfs://localhost:9000/user/root/modelNames/part-00000" spark.sparkContext.addFile(files) val path = SparkFiles.get("part-00000") println(path)打印的路径十分奇怪(没有截取完全): 然后有了path之后,就可以使用scala的I/O进行读取: val source = Source.fromFile(path) val lineIterator = source.getLines val lines =lineIterator.toArray println(lines.mkString(","))输出为: FM,FFM,DEEPFM,NFM,DIN,DIEN5、判断HDFS上文件路径是否存在在读取HDFS地址或者将文件传输到Driver上的时候,首先需要判断文件是否存在。单机环境下,代码如下: val conf = spark.sparkContext.hadoopConfiguration val path = new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames/part-00000") val fs = path.getFileSystem(conf) //得hdfs文件系统中的路径信息 val modelNamesExists = fs.exists(path) val modelNames1Exists = fs.exists(new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames1/part-00000")) println(modelNamesExists) println(modelNames1Exists)输出结果为: true false而在公司中的大规模集群环境下,通常的代码如下: val conf = spark.sparkContext.hadoopConfiguration val fs = org.apache.hadoop.fs.FileSystem.get(conf) val modelNamesExists = fs.exists(new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames/part-00000")) val modelNames1Exists = fs.exists(new org.apache.hadoop.fs.Path("hdfs://localhost:9000/user/root/modelNames1/part-00000")) println(modelNamesExists) println(modelNames1Exists)如果在本地单机环境下仍然使用上面的代码,会报如下的错误: Wrong FS: hdfs://localhost:9000/user/root/modelNames/part-00000, expected: file:///所以对比两份代码你可以发现,在本地环境中,我们首先使用getFileSystem获取了hdfs文件系统中的路径信息,从而避免了上面的错误。 好了,今天的知识就分享到这里,小伙伴们都掌握了么? |
今日新闻 |
推荐新闻 |
CopyRight 2018-2019 办公设备维修网 版权所有 豫ICP备15022753号-3 |